package my.thesis.bolts.computations;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SentimentDetectorPositive extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	String filePath="src/main/resources/AFINN-111-POSITIVE.txt";
	String fileName="AFINN-111-POSITIVE.txt";
		
	HashMap<String, Integer> AFFINWORDS = new HashMap<String,Integer>();
	Set<String> words= new HashSet<String>();
	
	private OutputCollector collector;
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("id","pos_score","country"));
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
	{
  	    this.collector=collector;
  	    
        BufferedReader bread = null;
        try 
        {				
				InputStream is = getClass().getClassLoader().getResourceAsStream(fileName);
            	bread = new BufferedReader(new InputStreamReader(is));				
		 } 
         catch (Exception  e) 
         {
        	 System.err.println(e.getMessage());
		 }
         
            String temp = "";
            
            try 
            {
                while ((temp = bread.readLine()) != null) 
                {
                    if (!temp.isEmpty()) 
                    {
                        String[] stwords = temp.split("\t");
                        String key = stwords[0];
                        Integer value =  Integer.parseInt(stwords[1]);
                        AFFINWORDS.put(key,value);  
                      }
                  }
            } 
            catch (IOException ex) 
            {
                
            }
            
	}
	
	public void execute(Tuple tuple) 
	{
		String country=tuple.getStringByField("country");
		Long id=tuple.getLongByField("id");
		
		ArrayList<String> words = (ArrayList<String>) tuple.getValueByField("words");
	
		int pos_score=0;
		
		//add the sentiment scores for every word in a tweet/ ignore words that are not present in dictionary (sentiment =0)		
		for(String w : words)
		{
			if(AFFINWORDS.containsKey(w))
			{
				pos_score=pos_score+AFFINWORDS.get(w);
			}
		}
		
		collector.emit(new Values(id,pos_score,country));  	
		
		//RELIABLE
		//collector.emit(tuple,new Values(pos_score,country));
		
		//RELIABLE
		//collector.ack(tuple);	
		
	}
}
